package gu.simplemq.activemq;

import java.util.Date;
import java.util.Map;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;

import com.google.common.collect.ImmutableMap;

import gu.simplemq.Channel;
import gu.simplemq.IMessageQueueFactory;
import gu.simplemq.IPublisher;
import gu.simplemq.jms.MessageQueueFactoryImpl;
import gu.simplemq.json.BaseJsonEncoder;

/**
 * @author guyadong
 *
 */
public class ActivemqPublisherTest implements ActivemqConstants {
    private static final int DELIVERY_MODE = DeliveryMode.NON_PERSISTENT;
	private static final String OPENWIRE_HOST = "192.168.10.226";

	private static ActiveMQConnectionFactory createFactory(){
		Properties props = new Properties();
    	props.setProperty("brokerURL","tcp://" + OPENWIRE_HOST + ":" + DEFAULT_OPENWIRE_PORT);
    	ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
       factory.setProperties(props);    
       return factory;
	}
	@Test
	public void test1() throws InterruptedException, JMSException {
		ActiveMQConnectionFactory factory = createFactory();
		Connection connection = null;
		Session session = null;
		MessageProducer p1 = null;
		MessageProducer p2 = null;
		try {
			connection = factory.createConnection("user", "password");
            connection.setExceptionListener(new MyExceptionListener());
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            p1 = session.createProducer(session.createTopic("chat1"));
            p2 = session.createProducer(session.createTopic("chat2"));
			for(int i=0; i<100; ++i){
				Date date = new Date();
				String str = "OPENWIRE "  + date.toString();
				String json = BaseJsonEncoder.getEncoder().toJsonString(str);
                TextMessage message = session.createTextMessage(json);
                p1.send(message, DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
                p2.send(message, DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);

				logger.info(date.getTime() +" : " +date.toString());
				Thread.sleep(2000);
			}
		} catch (JMSException e) {
			e.printStackTrace();
		} finally{
			if(null != p2){
				p2.close();
			}
			if(null != p1){
				p1.close();
			}
			if(null != session){
				session.close();
			}
			if(null != connection){
				connection.close();
			}
		}

	}
	@SuppressWarnings({ "rawtypes", "unchecked" })
	@Test
	public void test2() {
//		ImmutableMap<String,String> m = ImmutableMap.of(MQ_USERNAME,"user",MQ_PASSWORD,"password");
		ImmutableMap<String,String> m = ImmutableMap.of(MQ_URI, ACTIVEMQ_OPENWIRE_SCHEMA+"://user:password@" + OPENWIRE_HOST + ":"+DEFAULT_OPENWIRE_PORT);
		@SuppressWarnings("resource")
		IMessageQueueFactory factory = new MessageQueueFactoryImpl(ActivemqRuntimeContext.ACTIVEMQ_CONTEXT).init((Map)m);
		try{
			IPublisher publisher = factory.getPublisher();
			Channel<String> c1 = new Channel<String>("chat1"){};
			Channel<String> c2 = new Channel<String>("chat2"){};
			for(int i=0;i<100;++i){
				Date date = new Date();
				publisher.publish(c1, "OPENWIRE " + date.toString());
				publisher.publish(c2, "OPENWIRE " + date.toString());
				logger.info(date.getTime() +" : " +date.toString());
				logger.info("consumer count of chat1 {} consumer count of chat2 {}",
						publisher.getAdvisor().subscriberCountOf(c1.name),
						publisher.getAdvisor().subscriberCountOf(c2.name));
				Thread.sleep(2000);
			}
			factory.close();
		} catch (Throwable e) {
			e.printStackTrace();
		}
		
	}
    private static class MyExceptionListener implements ExceptionListener {
        @Override
        public void onException(JMSException exception) {
            System.out.println("Connection ExceptionListener fired, exiting.");
            exception.printStackTrace(System.out);
            System.exit(1);
        }
    }
}
